package yatan.cluster.concurrent;

import java.util.Date;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.log4j.Logger;

import yatan.cluster.job.Job;
import yatan.cluster.job.JobAuditor;
import yatan.cluster.job.JobResult;
import yatan.cluster.job.JobRunner;
import yatan.cluster.job.JobRunnerFactory;
import yatan.cluster.job.impl.DefaultJobAuditor;
import yatan.common.utility.LogUtility;

public class JobPool {
	private final Logger logger = Logger.getLogger(JobPool.class);

	private JobAuditor jobAuditor = new DefaultJobAuditor();

	private int threadPoolSize;

	private JobDoneHandler jobDoneHandler;
	private JobRunnerFactory jobRunnerFactory;

	private ExecutorService executorService;
	private int jobCount = 0;

	public JobPool(int size) {
		this.executorService = Executors.newFixedThreadPool(size);
		this.threadPoolSize = size;
	}

	public synchronized Future<JobResult> submit(Job job) {
		Future<JobResult> future = this.executorService
				.submit(new NotifiableTask(job));
		this.jobCount++;

		this.logger.debug("Submit job " + job + ", total job in pool is "
				+ this.jobCount);

		return future;
	}

	public synchronized int getJobCount() {
		return this.jobCount;
	}

	public synchronized void setJobDoneHandler(JobDoneHandler jobDoneHandler) {
		this.jobDoneHandler = jobDoneHandler;
	}

	public JobDoneHandler getJobDoneHandler() {
		return jobDoneHandler;
	}

	public void setJobRunnerFactory(JobRunnerFactory jobRunnerFactory) {
		this.jobRunnerFactory = jobRunnerFactory;
	}

	public JobRunnerFactory getJobRunnerFactory() {
		return jobRunnerFactory;
	}

	private class NotifiableTask implements Callable<JobResult> {
		private Job job;

		public NotifiableTask(Job job) {
			this.job = job;
		}

		public JobResult call() throws Exception {
			try {
				long start = new Date().getTime();

				JobRunner jobRunner = JobPool.this.jobRunnerFactory
						.createJobRunner(this.job.getRunnerName());
				if (jobRunner == null) {
					// FIXME: define an exception and throw it here
					throw new Exception("No runner named "
							+ this.job.getRunnerName());
				}

				JobPool.this.logger.debug("Starting job " + this.job);
				jobAuditor.audit(job, "start");

				JobResult jobResult;
				try {
					jobResult = jobRunner.runJob(this.job);
					jobResult.setId(this.job.getId());
				} catch (Exception e) {
					jobAuditor.audit(job, "fail");
					logger.debug("Job " + this.job + " exit with error.", e);
					if (JobPool.this.jobDoneHandler != null) {
						synchronized (JobPool.this.jobDoneHandler) {
							JobPool.this.jobDoneHandler.error(this.job, e);
						}
					}
					return null;
				} finally {
					synchronized (JobPool.this) {
						JobPool.this.jobCount--;
					}
				}

				jobAuditor.audit(job, "success");
				JobPool.this.logger
						.debug("Job " + this.job + " done! cost "
								+ ((double) new Date().getTime() - start)
								/ 1000 + " s");

				if (JobPool.this.jobDoneHandler != null) {
					synchronized (JobPool.this.jobDoneHandler) {
						JobPool.this.jobDoneHandler.success(jobResult);
					}
				}

				return jobResult;
			} catch (Exception e) {
				throw LogUtility.logError(logger, e);
			}
		}
	}
}
